package regular2MR;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Queue;

import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

public class EvalReducer extends
		Reducer<IntWritable, RvecWritable, IntWritable, BooleanWritable> {

	public void reduce(IntWritable key, Iterable<RvecWritable> values,
			Context context) throws IOException, InterruptedException {
		
		// get query
		Path[] fileCaches = DistributedCache.getLocalCacheFiles(context
				.getConfiguration());
		GraphQuery gq = Ultility.getAutomaton(fileCaches[0]);
		
		// RvecWritable total = new RvecWritable();
		ArrayList<Rvec> RVset = new ArrayList<Rvec>();
		Iterator<RvecWritable> itr = values.iterator();
		RvecWritable data = new RvecWritable();
		while (itr.hasNext()) {
			data = itr.next();
		}
		RVset.addAll(data.rvset);

		boolean answer = evalDG(RVset, gq);		
		
		BooleanWritable val = new BooleanWritable(answer);
		//System.out.println(answer);
		context.write(key, val);
	}

	public GraphQuery getAutomaton() {
		// construct the query automaton
		VertexQ vq0 = new VertexQ("4");
		VertexQ vq1 = new VertexQ("DB");
		VertexQ vq2 = new VertexQ("HR");
		VertexQ vq3 = new VertexQ("12");

		vq0.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq2.stateName)));
		vq1.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq3.stateName)));
		vq2.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq2.stateName), new EdgeQ(vq3.stateName)));

		ArrayList<VertexQ> Vq = new ArrayList<VertexQ>(Arrays.asList(vq0, vq1,
				vq2, vq3));
		GraphQuery gq = new GraphQuery(Vq, vq0, vq3);
		return gq;
	}

	public GraphQuery getAutomaton1() {
		// construct the query automaton
		VertexQ vq0 = new VertexQ("2124");
		VertexQ vq1 = new VertexQ("Film & Animation");
		VertexQ vq2 = new VertexQ("Entertainment");
		VertexQ vq3 = new VertexQ("949");

		vq0.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq2.stateName)));
		vq1.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq1.stateName), new EdgeQ(vq3.stateName)));
		vq2.childrenState = new ArrayList<EdgeQ>(Arrays.asList(new EdgeQ(
				vq2.stateName), new EdgeQ(vq3.stateName)));

		ArrayList<VertexQ> Vq = new ArrayList<VertexQ>(Arrays.asList(vq0, vq1,
				vq2, vq3));
		GraphQuery gq = new GraphQuery(Vq, vq0, vq3);
		return gq;
	}

	// A BFS based function to check whether s can reach t
	public static boolean isReachable(ArrayList<VertexD> Vd, VertexD s,
			VertexD t) {
		if (s.equals(t))
			return true;

		// Create a queue for BFS
		Queue<VertexD> queue = new LinkedList<VertexD>();
		s.visit = true;
		queue.add(s);

		while (!queue.isEmpty()) {
			VertexD v = queue.poll();

			for (EdgeD e : v.neighbors) {
				// If this neighbor is the target vertex, then return true
				if (e.target.equals(t))
					return true;
				// Else, continue to do BFS
				if (e.target.visit == false) {
					queue.add(e.target);
					e.target.visit = true;
				}
			}
		}
		return false;
	}

	public static boolean evalDG(ArrayList<Rvec> RVset, GraphQuery gq) {
		boolean answer = false;

		Vertex s = new Vertex(gq.us.stateName);
		Vertex t = new Vertex(gq.ut.stateName);

		// construct dependence graph from RVset
		ArrayList<VertexD> Vd = new ArrayList<VertexD>();
		String preVertex = "";
		for (Rvec r : RVset) {			
			if (!r.v.equals(preVertex)) {
				// - false
				if (r.value.get(0).value == true
						|| r.value.get(0).boolVar != null) {
					VertexD vd = new VertexD(r.v, r.u, r.value);
					Vd.add(vd);
					preVertex = r.v;
				}
			}
		}

		// Terminal state
		VertexD vdt = new VertexD(t.name, t.name, new ArrayList<BoolFormula>(
				Arrays.asList(new BoolFormula(true))));

		boolean checkVdt = false;
		ArrayList<VertexD> listRemove = new ArrayList<VertexD>();
		// set neighbors
		for (VertexD v : Vd) {			
			for (BoolFormula f : v.label) {				
				if (f.value == true) {
					if (!v.name.equals(vdt.name)) // final state in in-node
						v.neighbors.add(new EdgeD(vdt));
					else
						checkVdt = true;
				} else if (f.boolVar != null) {
					VertexD target = getVertexD(f.boolVar.v, Vd);
					if (target != null)
						v.neighbors.add(new EdgeD(target));
				}
			}			
			// if don't have neighbor then remove
			if (v.neighbors.size() == 0 && !v.name.equals(vdt.name))
				listRemove.add(v);

		}

		// remove all node don't match of any state in query automaton
		Vd.removeAll(listRemove);

		// Add terminal vertex
		if (!checkVdt)
			Vd.add(vdt);

		/*
		// print dependence graph
		for (VertexD vd : Vd) {
			System.out.print(vd.name + ": --> ");
			for (EdgeD e : vd.neighbors)
				System.out.print(e.target.name + " | ");
			System.out.print("\n");
		}
		*/

		// get start vertex in Vq
		VertexD vds = getVertexD(s.name, Vd);
		if (vds == null) {
			System.out.println("Result is false because start state had not appeared in dependence graph!");
			return false;
		}

		// BFS to get final answer
		answer = isReachable(Vd, vds, vdt);

		return answer;
	}

	public static VertexD getVertexD(String name, ArrayList<VertexD> Vd) {
		VertexD tmp = null;
		for (VertexD v : Vd)
			if (v.name.equals(name)) {
				tmp = v;
				break;
			}

		return tmp;
	}

}
